package com.hu.flink12.api.sql;

import com.hu.flink12.api.entity.Order;
import com.hu.flink12.api.source.CustomerOrder;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.Tumble;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.lit;

/**
 * @Author: hujianjun
 * @Date: 2021/2/9 23:02
 * @Describe: 使用FlinkSQl和Table来统计5s内 每个用户的订单总数、订单的最大金额、订单的最小金额（事件时间+wm+Flink sql和table的window完成统计）
 */
public class WindowSql {
    public static void main(String[] args) throws Exception {
        // TODO 1.获取env和tableEnv
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings envSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, envSettings);


        // TODO 3.将DataStream数据转为Table或View,然后查询
        DataStream<Order> inputDataStream = env.addSource(new CustomerOrder.CustomerOrderSource());

        //设置wm
        DataStream<Order> orderDataStream = inputDataStream.assignTimestampsAndWatermarks(WatermarkStrategy
                .<Order>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner((order, timestamp) -> order.getTimeStamp()));

        // 3.1 使用Sql查询    需要注意： $("timeStamp").rowtime()需要指定哪一列是时间
//        tableEnv.createTemporaryView("order_table", orderDataStream,
//                $("orderId").as("order_id"), $("userId").as("user_id"), $("money"), $("timeStamp").rowtime().as("time_stamp"));
//        String sql = "select user_id,count(order_id) as order_cnt,max(money) as max_money,min(money) as min_money from order_table group by user_id, tumble(time_stamp,interval '5' second)";
//        Table queryResult = tableEnv.sqlQuery(sql);

        // 3.2 使用Table API查询
        Table orderTable = tableEnv.fromDataStream(orderDataStream,
                $("orderId").as("order_id"), $("userId").as("user_id"), $("money"), $("timeStamp").rowtime().as("time_stamp"));
        Table queryResult = orderTable.window(Tumble.over(lit(5).second())
                .on($("time_stamp"))
                .as("tumble_window"))
                .groupBy($("tumble_window"), $("user_id"))
                .select(
                        $("user_id"),
                        $("order_id").count().as("order_cnt"),
                        $("money").max().as("max_money"),
                        $("money").min().as("min_money")
                );

        //打印schema信息
        queryResult.printSchema();

        //不能针对tableResult进行打印，需要转为DataStream才可以打印
        DataStream<Tuple2<Boolean, Row>> tuple2DataStream = tableEnv.toRetractStream(queryResult, Row.class);

        // TODO 4.sink
        tuple2DataStream.print();

        // TODO 5.执行
        env.execute();
    }
}
